package com.bds.snatch.system.mq;

import java.util.Date;
import java.util.Properties;

import org.apache.log4j.Logger;
import org.nutz.dao.Chain;
import org.nutz.dao.Cnd;
import org.nutz.dao.Dao;
import org.nutz.dao.impl.NutDao;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSONObject;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.PropertyValueConst;
import com.bds.snatch.model.H2Urls;
import com.bds.snatch.model.Task;
import com.bds.snatch.system.queue.TQueue;
import com.bds.snatch.system.util.SpringContextHolder;

@Service
public class SnatchMQConsumer {

	private static Logger logger = Logger.getLogger(SnatchMQConsumer.class);
	private static String accessKey = "hc44qMkvgMdfP51v";
	private static String secretKey = "T8rPyyqzj7RSfYM8ZOgRCYiARWxH1D";
	private static String task_consumerId = "CID_BDS_URL";
	private static String task_topic = "bds-url";
	private Dao h2dao;

	public void doStart(Dao h2dao) {
		this.h2dao = h2dao;
		consume();
	}

	public void consume() {

		try {
			NutDao apiDao = SpringContextHolder.getBean("dao");

			Properties properties = new Properties();
			properties.put(PropertyKeyConst.ConsumerId, task_consumerId);
			properties.put(PropertyKeyConst.AccessKey, accessKey);
			properties.put(PropertyKeyConst.SecretKey, secretKey);
			properties.put(PropertyKeyConst.SuspendTimeMillis, "100");
			// 消息消费失败时的最大重试次数
			properties.put(PropertyKeyConst.MaxReconsumeTimes, "20");
			properties.put(PropertyKeyConst.ONSAddr,
					"http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet");
			properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
			Consumer consumer = ONSFactory.createConsumer(properties);
			consumer.subscribe(task_topic, "*", new MessageListener() {
				public Action consume(Message message, ConsumeContext context) {
					String body = new String(message.getBody());
					logger.info(">>>>Receive: " + body);
					try {

						JSONObject msg = JSONObject.parseObject(body);
						String table = msg.getString("table");
						Long id = msg.getLong("id");
						String url = msg.getString("url");
						String isjs = msg.getString("isjs");
						String requestId = msg.getString("requestId");
						Long user_id = msg.getLong("userId");
						if (table.equals("t_template_url")) {

							H2Urls urls = new H2Urls();
							urls.setId(id);
							urls.setUser_id(user_id);
							urls.setRequest_id(requestId);
							urls.setUrl(url);
							urls.setIsjs(isjs);
							urls.setCreate_time(new Date());
							urls.setType("url");
							TQueue.getQueue().add(urls);
							h2dao.fastInsert(urls);
						} else if (table.equals("t_task")) {
							H2Urls urls = new H2Urls();
							urls.setId(id);
							urls.setUser_id(user_id);
							urls.setRequest_id(requestId);
							urls.setUrl(url);
							urls.setIsjs(isjs);
							urls.setCreate_time(new Date());
							urls.setType("portal");
							TQueue.getQueue().add(urls);
							apiDao.update(Task.class, Chain.make("status", 1), Cnd.where("id", "=", id));
						}

					} catch (Exception e) {
						logger.error("", e);

					}

					return Action.CommitMessage;
				}
			});
			consumer.start();
			logger.info(">>>>开始监听消息同步任务队列, ConsumerId:" + task_consumerId);
		} catch (Exception e) {
			logger.error("", e);
		}

	}

}
